package com.hujf.flink.mysql;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * @Author hujf
 * @Date 2022/5/11 10:21
 * @Description TODO
 */
public class MysqlTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
                .hostname("10.0.6.42")
                .port(3306)
                .username("root")
                .password("Rxsk@2022")
                .databaseList("schedule")
                    //开始时间 最新
                .startupOptions(StartupOptions.latest())
                .deserializer(new DebeziumDeserializationSchema<String>() {
                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }

                    @Override
                    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
                        String topic = sourceRecord.topic();
                        collector.collect(topic);
                    }
                })
                .build();

        environment.addSource(mysqlSource).print().setParallelism(1);


        environment.execute();
//        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
//                .hostname("10.0.6.42")
//                .port(3306)
//                .username("root")
//                .password("Rxsk@2022")
//                .databaseList("schedule")
//                .deserializer(new StringDebeziumDeserializationSchema())
//                // converts SourceRecord to String
//                .build();
//
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
//
//        env.addSource(sourceFunction).print().setParallelism(1);
//        // use parallelism 1 for sink to keep message ordering
//        env.execute("test");
    }
}
